/**
 * @file   node.cpp
 * @author sunway <sunwayforever@gmail.com>
 * @date   Tue Sep 23 15:30:35 2008
 * 
 * @brief  
 * 
 * 
 */

#include "node.h"
extern "C" {
#include "wrapper.h"
}
using namespace DHT;
ostream & DHT::operator<<(ostream & o,const NodeId & p) {
    for(int i = 0; i < IDLEN; i++) {
	printf("%02x", p.digest[i]);
    }
    return o;
}

NodeId  NodeId::operator^ (const NodeId & r) const {
    NodeId tmp;
    for (int i=0;i<IDLEN;i++) {
	tmp.digest[i]=digest[i]^r.digest[i];
    }
    return tmp;
}

/** 
 * determine which bucket should the nodeid be placed
 * 
 * 
 * @return bucket index
 */
int NodeId::getBucket () const {
    for (int i=0; i<IDLEN; ++i) {
	for (int j=7; j>=0; --j) {
	    if (digest[i]&1<<j) {
		return 7-j+8*i;
	    }
	}
    }
}

void * DHT::republishCallBack (void * p) {
    Node * np=(Node *)p;
    while (1) {
	sleep (tRepublish);
	for (vector<Cache>::iterator ite=np->cachePut.begin ();ite!=np->cachePut.end ();++ite) {
	    np->put (ite->key,ite->buff,ite->length);
	}
    }
}

void * DHT::refreshCallBack (void * p) {
    Node * np=(Node *)p;
    while (1) {
	sleep (tRefresh);
	PredRemove predRemove(np);
	pthread_mutex_lock (&np->kBucketMutex);
	for (int i=0; i<IDLEN*8; ++i) {
	    np->kBucket[i].erase (remove_if (np->kBucket[i].begin (),np->kBucket[i].end (),predRemove),np->kBucket[i].end ());
	}
	pthread_mutex_unlock (&np->kBucketMutex);
    }
}

void * DHT::listenCallBack (void * p) {
    Node * np=(Node *)p;
#ifdef _USE_IPV6
    struct sockaddr_in6 servaddr;
#else
    struct sockaddr_in servaddr;
#endif
    bzero ((void *)&servaddr,sizeof(servaddr));
#ifdef _USE_TCP
#ifdef _USE_IPV6
    int listenfd=Socket (PF_INET6,SOCK_STREAM,0);
    servaddr.sin6_family=PF_INET6;
    servaddr.sin6_addr=in6addr_any;
    servaddr.sin6_port=htons (np->myTuple.port);
#else 
    int listenfd=Socket (AF_INET,SOCK_STREAM,0);
    servaddr.sin_family=AF_INET;
    servaddr.sin_addr.s_addr=htonl (INADDR_ANY);
    servaddr.sin_port=htons (np->myTuple.port);
#endif
    if (Bind (listenfd, (struct sockaddr *)&servaddr,sizeof(servaddr))) {
	cout<<"err when Bind"<<endl;
	exit (1);
    }
    Listen (listenfd,0);
#else
#ifdef _USE_IPV6
    int sockfd=Socket (PF_INET6,SOCK_DGRAM,0);
    servaddr.sin6_family=PF_INET6;
    servaddr.sin6_addr=in6addr_any;
    servaddr.sin6_port=htons (np->myTuple.port);
#else
    int sockfd=Socket (AF_INET,SOCK_DGRAM,0);
    servaddr.sin_family=AF_INET;
    servaddr.sin_addr.s_addr=htonl (INADDR_ANY);
    servaddr.sin_port=htons (np->myTuple.port);
#endif
    Bind (sockfd, (struct sockaddr *)&servaddr,sizeof(servaddr));
#endif
    char inPktBuff[PKTSIZE]={};
    while (1) {
#ifdef _USE_IPV6
	struct sockaddr_in6 fromaddr;
#else
	struct sockaddr_in fromaddr;
#endif
	bzero (&fromaddr,sizeof(fromaddr));
	socklen_t fromAddrLen=sizeof(fromaddr);
#ifdef _USE_TCP
	int sockfd=accept (listenfd,(struct sockaddr *)&fromaddr,&fromAddrLen);
	read (sockfd,inPktBuff,PKTSIZE);
#else
	Recvfrom(sockfd,inPktBuff,PKTSIZE,0,(struct sockaddr *)&fromaddr,&fromAddrLen);
#endif
	char type=inPktBuff[0];
	if (type==TYPE_PING) {
	    // format:type[8 bits]  nodeId [160 bits] port [16 bits] 23 bytes total
	    char from_ip[IPSTRLEN]={};
#ifdef _USE_IPV6
	    inet_ntop (PF_INET6,&fromaddr.sin6_addr,from_ip,IPSTRLEN);
#else
	    inet_ntop (AF_INET,&fromaddr.sin_addr,from_ip,IPSTRLEN);
#endif
	    unsigned char tmp_id[IDLEN]={};
	    memcpy (tmp_id,inPktBuff+1,IDLEN);
	    uint16_t from_port=0;
	    memcpy (&from_port,inPktBuff+1+IDLEN,sizeof(from_port));
	    char outPktBuff[1];
	    outPktBuff[0]=TYPE_PONG;
#ifdef _USE_TCP
	    write (sockfd,outPktBuff,1);
	    close (sockfd);
#else
	    Sendto (sockfd,outPktBuff,1,0,(struct sockaddr *)&fromaddr,fromAddrLen);
#endif
	    np->localRegNode (NodeId (tmp_id),string (from_ip),from_port);
	} else if (type==TYPE_FIND_NODE) {
	    unsigned char tmp_id[IDLEN]={};
	    memcpy (tmp_id,inPktBuff+1,IDLEN);
	    vector<Tuple> ret=np->localFindNode (NodeId (tmp_id),K);
	    bzero (tmp_id,IDLEN);
	    memcpy (tmp_id,inPktBuff+IDLEN+1,IDLEN);
	    uint16_t tmp_port=0;
	    memcpy (&tmp_port,inPktBuff+1+IDLEN+IDLEN,sizeof(tmp_port));
	    char tmp_ip[IPSTRLEN]={};
#ifdef _USE_IPV6
	    inet_ntop (PF_INET6,&fromaddr.sin6_addr,tmp_ip,IPSTRLEN);
#else
	    inet_ntop (AF_INET,&fromaddr.sin_addr,tmp_ip,IPSTRLEN);
#endif
	    np->localRegNode (NodeId (tmp_id),string (tmp_ip),tmp_port);
	    // format: type[8 bits] N[0-20,8 bits] nodeid1 [160 bits] ip1 [32
	    // bits] port1 [16 bits], 560 bytes at most
	    char outPktBuff[PKTSIZE];
	    outPktBuff[0]=TYPE_RESPONSE_FIND_NODE;
	    outPktBuff[1]=ret.size ();
	    int curr_offset=2;
	    for (vector<Tuple>::iterator ite=ret.begin (); ite!=ret.end (); ++ite) {
		const unsigned char * tmp_id=ite->nodeId.getVal ();
#ifdef _USE_IPV6
		struct sockaddr_in6 servaddr;
		bzero(&servaddr, sizeof(servaddr));
		servaddr.sin6_family = PF_INET6;
		inet_pton(PF_INET6, ite->ip.c_str (), &servaddr.sin6_addr);
		memcpy (outPktBuff+curr_offset,tmp_id,IDLEN);
		memcpy (outPktBuff+curr_offset+IDLEN,&servaddr.sin6_addr,IPLEN);
		memcpy (outPktBuff+curr_offset+IDLEN+IPLEN,&ite->port,sizeof(ite->port));
		curr_offset+=IDLEN+IPLEN+sizeof(ite->port);
#else
		struct sockaddr_in servaddr;
		bzero(&servaddr, sizeof(servaddr));
		servaddr.sin_family = AF_INET;
		inet_pton(AF_INET, ite->ip.c_str (), &servaddr.sin_addr.s_addr);
		memcpy (outPktBuff+curr_offset,tmp_id,IDLEN);
		memcpy (outPktBuff+curr_offset+IDLEN,&servaddr.sin_addr.s_addr,IPLEN);
		memcpy (outPktBuff+curr_offset+IDLEN+IPLEN,&ite->port,sizeof(ite->port));
		curr_offset+=IDLEN+IPLEN+sizeof(ite->port);
#endif
	    }
#ifdef _USE_TCP	    
	    write (sockfd,outPktBuff,curr_offset);
	    close (sockfd);
#else
	    Sendto (sockfd,outPktBuff,curr_offset,0,(struct sockaddr *)&fromaddr,fromAddrLen);
#endif
	} else if (type==TYPE_QUIT) {
	    // format: type[8 bits] nodeid[160 bits] 21 bytes total
	    unsigned char tmp_id[IDLEN]={};
	    memcpy (tmp_id,inPktBuff+1,IDLEN);
	    np->quit (NodeId (tmp_id));
	    np->localDelNode (NodeId (tmp_id));
	} else if (type==TYPE_STORE) {
	    // format:type[8 bits]  nodeId [160 bits] data size [16 bits]
	    // data [65535 bytes at most] 23+n bytes total
	    unsigned char tmpId[IDLEN]={0};
	    memcpy (tmpId,inPktBuff+1,IDLEN);
	    uint16_t tmpLen=0;
	    memcpy (&tmpLen,inPktBuff+1+IDLEN,sizeof(tmpLen));
	    char tmpBuff[tmpLen];
	    bzero (tmpBuff,tmpLen);
	    memcpy (tmpBuff,inPktBuff+1+IDLEN+sizeof(tmpLen),tmpLen);
	    vector<Cache>::iterator ite=find (np->cacheGet.begin (),np->cacheGet.end (),Cache (NodeId (tmpId)));
	    if (ite!=np->cacheGet.end ()) { // cache already exist, so replace it
		np->cacheGet.erase (ite);
	    }
	    np->cacheGet.push_back (Cache (NodeId (tmpId),tmpBuff,tmpLen));
	    cout<<"STORE recved for nodeId "<<NodeId (tmpId)<<"and len is "<<np->cacheGet.back ().length<<endl;
	} else if (type==TYPE_FETCH) {
	    // format:type[8 bits]  nodeId [160 bits] data size [16 bits] 23 bytes total
	    // data [65535 bytes at most] 23+n bytes total
	    unsigned char tmpId[IDLEN]={0};
	    memcpy (tmpId,inPktBuff+1,IDLEN);
	    uint16_t tmpLen=0;
	    memcpy (&tmpLen,inPktBuff+1+IDLEN,sizeof(tmpLen));
	    vector<Cache>::iterator ite=find (np->cacheGet.begin (),np->cacheGet.end (),Cache (NodeId (tmpId)));
	    const char * dataBuff;
	    uint16_t dataLen;
	    if (ite!=np->cacheGet.end ()) {
		if (time (0)>ite->deadLine) { // cache is expired
		    dataBuff=0;
		    dataLen=0;
		    np->cacheGet.erase (ite);		    
		} else {
		    dataBuff=ite->buff;
		    dataLen=ite->length;
		}
	    } else {		// unlikely
		dataBuff=0;
		dataLen=0;
	    }
	    cout<<"FETCH recved for NodeId "<<NodeId (tmpId)<<" and len is "<<dataLen<<endl;
	    // format:type[8 bits] data size [16 bits] data [65535 bytes at most] 3+n bytes total
	    int outPktLen=dataLen+sizeof(dataLen)+1;
	    char outPktBuff[outPktLen];
	    bzero (outPktBuff,outPktLen);
	    outPktBuff[0]=TYPE_RESPONSE_FETCH;
	    memcpy (outPktBuff+1,&dataLen,sizeof(dataLen));
	    memcpy (outPktBuff+1+sizeof(dataLen),dataBuff,dataLen);
#ifdef _USE_TCP
	    write (sockfd,outPktBuff,outPktLen);
	    close (sockfd);
#else
	    Sendto (sockfd,outPktBuff,outPktLen,0,(struct sockaddr *)&fromaddr,fromAddrLen);
#endif
	}
    }
}

void Node::localDelNode (NodeId id) {
    // todo
    int index=id.getBucket ();
    kBucket[index].remove (Tuple (id));
    cout<<"node "<<id<<" quited"<<endl;
}

void Node::quit () {
    vector<Tuple> contactToNotify=localFindNode (myTuple.nodeId,K);
    for (vector<Tuple>::iterator ite=contactToNotify.begin (); ite!=contactToNotify.end (); ++ite) {
	// format:type[8 bits]  nodeId [160 bits] 21 bytes total
	char outPktBuff[1+IDLEN]={0};
	outPktBuff[0]=TYPE_QUIT;
	memcpy (outPktBuff+1,myTuple.nodeId.getVal (),IDLEN);
#ifdef _USE_IPV6
	struct sockaddr_in6 servaddr;
	bzero(&servaddr, sizeof(servaddr));
	servaddr.sin6_family = PF_INET6;
	servaddr.sin6_port = htons(ite->port);
	inet_pton(PF_INET6, ite->ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
	int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
	connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
	write (sockfd,outPktBuff,1+IDLEN);
#else
	int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
	Sendto(sockfd, outPktBuff, 1+IDLEN, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
	struct sockaddr_in servaddr;
	bzero(&servaddr, sizeof(servaddr));
	servaddr.sin_family = AF_INET;
	servaddr.sin_port = htons(ite->port);
	inet_pton(AF_INET, ite->ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
	int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
	connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
	write (sockfd,outPktBuff,1+IDLEN);
#else
	int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
	Sendto(sockfd, outPktBuff, 1+IDLEN, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
	close (sockfd);
    }
    
}

void Node::quit (NodeId id) {
    int index=id.getBucket ();
    if (find (kBucket[index].begin (),kBucket[index].end (),Tuple (id))==kBucket[index].end ()) {
	return;			// return if id is not in my contact list
    }
    vector<Tuple> contactToNotify=localFindNode (id,K);
    for (vector<Tuple>::iterator ite=contactToNotify.begin (); ite!=contactToNotify.end (); ++ite) {
	// format:type[8 bits]  nodeId [160 bits] 21 bytes total
	char outPktBuff[1+IDLEN]={0};
	outPktBuff[0]=TYPE_QUIT;
	memcpy (outPktBuff+1,id.getVal (),IDLEN);
#ifdef _USE_IPV6
	struct sockaddr_in6 servaddr;
	bzero(&servaddr, sizeof(servaddr));
	servaddr.sin6_family = PF_INET6;
	servaddr.sin6_port = htons(ite->port);
	inet_pton(PF_INET6, ite->ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
	int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
	connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
	write (sockfd,outPktBuff,1+IDLEN);
#else
	int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
	Sendto(sockfd, outPktBuff, 1+IDLEN, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
	struct sockaddr_in servaddr;
	bzero(&servaddr, sizeof(servaddr));
	servaddr.sin_family = AF_INET;
	servaddr.sin_port = htons(ite->port);
	inet_pton(AF_INET, ite->ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
	int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
	connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
	write (sockfd,outPktBuff,1+IDLEN);
#else
	int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
	Sendto(sockfd, outPktBuff, 1+IDLEN, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
	close (sockfd);
    }
    
}

void Node::bootsTrap (const string & ip,uint16_t port) {
    if (ip==myTuple.ip && port==myTuple.port) {
	return;
    }
    cout<<"connecting to "<<ip<<":"<<port<<" for bootstrap"<<endl;
    // todo:error
    vector<Tuple> contact=FIND_NODE (myTuple.nodeId,ip,port);
    vector<Tuple> tmp_contact;
    while (!contact.empty ()) {
	tmp_contact.clear ();
	for (vector<Tuple>::iterator ite=contact.begin (); ite!=contact.end (); ++ite) {
	    if (!localRegNode (ite->nodeId,ite->ip,ite->port)) {
		vector<Tuple> tmp=FIND_NODE (myTuple.nodeId,ite->ip,ite->port);		
		for (vector<Tuple>::iterator ite2=tmp.begin (); ite2!=tmp.end (); ++ite2) {
		    tmp_contact.push_back (*ite2);		    
		}
	    }
	}
	contact=tmp_contact;
    }
    cout<<"bootstrap done"<<endl;
}

vector<Tuple> Node::lookUp (NodeId nodeId) const {
    vector<Tuple> shortList=localFindNode (nodeId,ALPHA);
    if (shortList.empty ()) {
	shortList.push_back (myTuple);
	return shortList;
    } 
    sort (shortList.begin (),shortList.end (),SortCompareFn (nodeId));
    vector<Tuple> validShortList; // will be returned as K closest tuples to nodeId
    vector<Tuple> tmpShortList;
    vector<Tuple> oldShortList;	
    while (validShortList.size ()<K+1) {
	tmpShortList.clear ();
	int count=0;
	while (count!=ALPHA && (!shortList.empty ())) {
	    vector<Tuple> ret;
	    const Tuple current=shortList.back ();
	    try {
		count++;
		shortList.pop_back ();
		oldShortList.push_back (current);
		ret=FIND_NODE (nodeId,current.ip,current.port);
	    } catch (const ExTimeout & e) {
		cerr<<e.what ()<<endl;
		continue;
	    }
	    validShortList.push_back (current);
	    for (vector<Tuple>::iterator ite=ret.begin ();ite!=ret.end ();++ite) {
		tmpShortList.push_back (*ite);
	    }
	}
	const Tuple & closest=shortList.back ();
	for (vector<Tuple>::iterator ite=tmpShortList.begin ();ite!=tmpShortList.end ();++ite) {
	    if (
		find (oldShortList.begin (),oldShortList.end (),*ite)==oldShortList.end () &&
		find (shortList.begin (),shortList.end (),*ite)==shortList.end () 
		)
	    {

		shortList.push_back (*ite);
	    }
	}
	sort (shortList.begin (),shortList.end (),SortCompareFn (nodeId));
	if (shortList.back ()==closest) {
	    // means new added tuples is not closer
	    break;
	}
	
    }
    assert (!validShortList.empty ());
    sort (validShortList.begin (),validShortList.end (),SortCompareFn (nodeId));
    reverse (validShortList.begin (),validShortList.end ());
    return validShortList;

}

/** 
  * register a node to kBucket
  * 
  * @param nodeid 
  * @param ip 
  * @param port 
  * 
  * @return 0 - register succ (node is new)  1 - register fail (node is old)
  */
int Node::localRegNode (const NodeId & nodeid,const string & ip,uint16_t port) {
    // diff nodeid and this->myTuple.nodeId to get an k-bucket index ranged from
    // 0-159
    if (ip==myTuple.ip && port==myTuple.port) {
	return 1;
    }
    pthread_mutex_lock (&kBucketMutex);
    int index=(nodeid^(myTuple.nodeId)).getBucket ();
    Tuple tmp_tuple (nodeid,ip,port);
    list<Tuple>::iterator ite=find (kBucket[index].begin (),kBucket[index].end (),tmp_tuple);
    int succ=0;
    if (ite==kBucket[index].end ()) {
	// contact doesn't exist in the bucket yet
	if (kBucket[index].size ()==K) {
	    // kBucket has K contact already
	    Tuple tmp=kBucket[index].front ();
	    if (PING (tmp.ip,tmp.port)) { // contact is dead
		kBucket[index].pop_front ();
		kBucket[index].push_back (tmp_tuple);
//		cout<<"registered:"<<nodeid<<" "<<ip<<" "<<port<<" to bucket "<<index<<endl;
		succ=0;
	    } else {		// contact is still alive
		kBucket[index].pop_front ();
		kBucket[index].push_back (tmp);
		succ=1;
	    }
	} else {
	    kBucket[index].push_back (tmp_tuple);
//	    cout<<"registered:"<<nodeid<<" "<<ip<<" "<<port<<" to bucket "<<index<<endl;
	    succ=0;
	}
    } else {
	// contact is already in the bucket
	// move the contact to the end of the list
	kBucket[index].push_back (*ite);
	kBucket[index].erase (ite);
//	cout<<"updated nodeinfo: "<<nodeid<<" "<<ip<<" "<<port<<" to bucket "<<index<<endl;
	succ=1;
    }
    pthread_mutex_unlock (&kBucketMutex);
    return succ;
}

/** 
 * tell [ip]:[port] to store a bulk of data 
 * 
 * @param ip 
 * @param port 
 * @param buff 
 * @param size 
 * 
 * @return 
 */
int Node::STORE (const string & ip,uint16_t port,const NodeId & key,const char * dataBuff,uint16_t nbyte) const {
    // format:type[8 bits]  nodeId [160 bits] data size [16 bits]
    // data [65535 bytes at most] 23+n bytes total
    uint16_t outPktLen=1+IDLEN+sizeof(nbyte)+nbyte;
    char outPktBuff[outPktLen];
    bzero (outPktBuff,outPktLen);    
    outPktBuff[0]=TYPE_STORE;
    memcpy (outPktBuff+1,key.getVal (),IDLEN);
    memcpy (outPktBuff+1+IDLEN,&nbyte,sizeof(nbyte));
    memcpy (outPktBuff+1+IDLEN+sizeof(nbyte),dataBuff,nbyte);
#ifdef _USE_IPV6
    struct sockaddr_in6 servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin6_family = PF_INET6;
    servaddr.sin6_port = htons(port);
    inet_pton(PF_INET6, ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
    int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
    struct sockaddr_in servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
    int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
    close (sockfd);
    // todo:recv the answer to determine whether store if fullfilled
    return 0;
}

int Node::FIND_VALUE (const string & ip,uint16_t port,const NodeId & key,char * dataBuff,uint16_t nbyte) const {
    // format:type[8 bits]  nodeId [160 bits] data size [16 bits] 23 bytes total
    // todo
    const int outPktLen=1+IDLEN+sizeof(nbyte);
    char outPktBuff[outPktLen]={};
    outPktBuff[0]=TYPE_FETCH;
    memcpy (outPktBuff+1,key.getVal (),IDLEN);
    memcpy (outPktBuff+1+IDLEN,&nbyte,sizeof(nbyte));
#ifdef _USE_IPV6
    struct sockaddr_in6 servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin6_family = PF_INET6;
    servaddr.sin6_port = htons(port);
    inet_pton(PF_INET6, ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
    int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
    struct sockaddr_in servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
    int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
    int ret=0;
    if (!WAIT (sockfd)) {
	char inPktBuff[PKTSIZE]={0};
#ifdef _USE_TCP
	read (sockfd,inPktBuff,PKTSIZE);
#else
	Recvfrom(sockfd,inPktBuff,PKTSIZE,0,NULL,NULL);
#endif
	// format:type[8 bits] data size [16 bits] data [65535 bytes at most] 3+n bytes total	
	char type=inPktBuff[0];
	if (type==TYPE_RESPONSE_FETCH) {
	    int tmp_len=0;
	    memcpy (&tmp_len,inPktBuff+1,2);
	    ret=tmp_len>nbyte?nbyte:tmp_len;
	    memcpy (dataBuff,inPktBuff+1+2,ret);
	}
    } else {
	cout<<"timeout when fetching"<<endl;
    }
    close (sockfd);
    return ret;
}
    
int Node::PING (const string & ip,uint16_t port) const {
    // format:type[8 bits]  nodeId [160 bits] port [16 bits] 23 bytes total
    const int outPktLen=1+sizeof(port)+IDLEN;
    char outPktBuff[outPktLen]={};
    outPktBuff[0]=TYPE_PING;
    memcpy (outPktBuff+1,myTuple.nodeId.getVal (),IDLEN);
    memcpy (outPktBuff+1+IDLEN,&myTuple.port,sizeof(port));
#ifdef _USE_IPV6    
    struct sockaddr_in6 servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin6_family = PF_INET6;
    servaddr.sin6_port = htons(port);
    inet_pton(AF_INET, ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
    int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
    struct sockaddr_in servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
    int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
    if (!WAIT(sockfd)) {
	char inPktBuff[PKTSIZE]={0};
#ifdef _USE_TCP
	read (sockfd,inPktBuff,PKTSIZE);
#else
	Recvfrom(sockfd,inPktBuff,PKTSIZE,0,NULL,NULL);
#endif
	char type=inPktBuff[0];
	if (type==TYPE_PONG) {
	    close (sockfd);
	    return 0;
	}
    }
    close (sockfd);
    return 1;
}

/** 
 * public key:data pair to the network
 * 
 * @param nodeId key
 * @param data data
 * @param nbyte nbyte of data
 * 
 * @return 0 - succ 1 - fail
 */
int Node::put (const NodeId & nodeId,const char * data,uint16_t nbyte ) {
    vector<Cache>::iterator ite=find (cachePut.begin (),cachePut.end (),Cache (nodeId));
    if (ite!=cachePut.end ()) {
	cachePut.erase (ite);
    }
    cachePut.push_back (Cache (nodeId,data,nbyte));
    vector<Tuple> dest=lookUp (nodeId);
    for (vector<Tuple>::iterator ite=dest.begin ();ite!=dest.end ();++ite) {
	STORE (ite->ip,ite->port,nodeId,data,nbyte);
    }
}

/** 
 * get data assoc with key from the network
 * 
 * @param nodeId key
 * @param data store the ret data, should be maintained by user
 * @param nbyte nbyte to get
 * 
 * @return 0 - fail >0 - size of data
 */
int Node::get (const NodeId & nodeId,char * data,int nbyte) {
    vector<Tuple> dest=lookUp (nodeId);
    int ret=0;
    for (vector<Tuple>::iterator ite=dest.begin ();ite!=dest.end ();++ite) {
	cout<<"requesting "<<ite->port<<endl;
	ret=FIND_VALUE (ite->ip,ite->port,nodeId,data,nbyte);
	assert (ret>=0);
	if (ret>0) {
	    break;
	}
    }
    // todo: store ret to those in dest who doesn't have the value
    return ret;
}

/** 
 * find K nodes closest to [nodeid] from [ip]:[port]'s contact list
 * 
 * @param nodeid 
 * @param ip 
 * @param port 
 * 
 * @return 
 */
vector<Tuple> Node::FIND_NODE (NodeId nodeId,const string & ip,uint16_t port) const throw (ExTimeout) {
    // format: type [8 bits] nodidToFind [160 bits] nodeid[160 bits]  port [16 bits] 43 bytes total
    const int outPktLen=1+IDLEN+IDLEN+sizeof(port);
    char outPktBuff[outPktLen]={};
    outPktBuff[0]=TYPE_FIND_NODE;
    memcpy (outPktBuff+1,nodeId.getVal (),IDLEN);
    memcpy (outPktBuff+1+IDLEN,myTuple.nodeId.getVal (),IDLEN);
    memcpy (outPktBuff+1+IDLEN+IDLEN,&myTuple.port,sizeof(port));
#ifdef _USE_IPV6
    struct sockaddr_in6 servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin6_family = PF_INET6;
    servaddr.sin6_port = htons(port);
    inet_pton(PF_INET6, ip.c_str (), &servaddr.sin6_addr);
#ifdef _USE_TCP
    int sockfd = Socket (PF_INET6, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (PF_INET6, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#else
    struct sockaddr_in servaddr;
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(port);
    inet_pton(AF_INET, ip.c_str (), &servaddr.sin_addr);
#ifdef _USE_TCP
    int sockfd = Socket (AF_INET, SOCK_STREAM, 0);
    connect (sockfd,(struct sockaddr *)&servaddr,sizeof(servaddr));
    write (sockfd,outPktBuff,outPktLen);
#else
    int sockfd = Socket (AF_INET, SOCK_DGRAM, 0);
    Sendto(sockfd, outPktBuff, outPktLen, 0, (struct sockaddr *)&servaddr, sizeof(servaddr));
#endif
#endif
    if (!WAIT(sockfd)) {
	char inPktBuff[PKTSIZE]={0};
#ifdef _USE_TCP
	read (sockfd,inPktBuff,PKTSIZE);
#else
	Recvfrom(sockfd,inPktBuff,PKTSIZE,0,NULL,NULL);
#endif
	char type=inPktBuff[0];
	if (type==TYPE_RESPONSE_FIND_NODE) {
	    int count=inPktBuff[1];
	    vector<Tuple> ret;
	    for (int i=0; i<count; ++i) {
		unsigned char tmp_id[IDLEN]={};
#ifdef _USE_IPV6
		memcpy (tmp_id,inPktBuff+2+i*(IDLEN+IPLEN+2),IDLEN);
		char tmp_ip[IPLEN]={};
		memcpy (tmp_ip,inPktBuff+2+20+i*(IDLEN+IPLEN+2),IPLEN);
		uint16_t tmp_port=0;
		memcpy (&tmp_port,inPktBuff+2+IDLEN+IPLEN+i*(IDLEN+IPLEN+2),sizeof(tmp_port));

		struct sockaddr_in6 servaddr;
		bzero(&servaddr, sizeof(servaddr));
		servaddr.sin6_family = PF_INET6;
		memcpy (&servaddr.sin6_addr,tmp_ip,IPLEN);
		char ip[IPSTRLEN]={0};
		inet_ntop (AF_INET6,&servaddr.sin6_addr,ip,IPSTRLEN);
		ret.push_back (Tuple (NodeId (tmp_id),string (ip),tmp_port));
#else
		memcpy (tmp_id,inPktBuff+2+i*(IPLEN+IDLEN+2),IDLEN);
		char tmp_ip[IPLEN]={};
		memcpy (&tmp_ip,inPktBuff+2+IDLEN+i*(IDLEN+IPLEN+2),IPLEN);
		uint16_t tmp_port=0;
		memcpy (&tmp_port,inPktBuff+2+IDLEN+IPLEN+i*(IDLEN+IPLEN+2),sizeof(tmp_port));

		struct sockaddr_in servaddr;
		bzero(&servaddr, sizeof(servaddr));
		servaddr.sin_family = AF_INET;
		memcpy (&servaddr.sin_addr.s_addr,tmp_ip,IPLEN);
		char ip[IPSTRLEN]={0};
		inet_ntop (AF_INET,&servaddr.sin_addr,ip,IPSTRLEN);
		ret.push_back (Tuple (NodeId (tmp_id),string (ip),tmp_port));
#endif
	    }
	    close (sockfd);
	    //cout<<"size: "<<ret.size ()<<" "<<count<<endl;
	    return ret;    
	}
    } else {
	close (sockfd);
	throw ExTimeout ();
    }
}

/** 
 * find [n] nodes closest to [nodeid] from local contact list
 * 
 * @param nodeid 
 * @param n 
 * 
 * @return 
 */
vector<Tuple> Node::localFindNode (NodeId nodeid,int n) const {
    vector<Tuple> ret;
    int index=(myTuple.nodeId^nodeid).getBucket ();
    list<Tuple>::const_iterator ite=kBucket[index].begin ();
    int curr_size=0;
    while (ite!=kBucket[index].end () && (++curr_size)<n+1) {
	ret.push_back (*ite);
	++ite;
    }
    if (curr_size==(n+1)) {
	return ret;
    } else {
	// todo
	int front=index-1;
	int back=index+1;
	int last=n-curr_size+1;
	assert (last>=1);
	while (front!=-1 || back!=IDLEN*8) {
	    if (back!=IDLEN*8) {
		for (list<Tuple>::const_iterator ite=kBucket[back].begin (); ite!=kBucket[back].end (); ++ite) {
		    ret.push_back (*ite);
		    if ((--last)==0) {
			break;
		    }
		}
		back++;
	    }
	    if (last==0) {
		break;
	    }
	    if (front!=-1) {
		for (list<Tuple>::const_iterator ite=kBucket[front].begin (); ite!=kBucket[front].end (); ++ite) {
		    ret.push_back (*ite);
		    if ((--last)==0) {
			break;
		    }
		}
		front--;
	    }
	    if (last==0) {
		break;
	    }
	}
	return ret;
    }
}

bool DHT::WAIT (int sockfd) {
    struct timeval timeout;
    timeout.tv_sec=TIMEOUT;
    timeout.tv_usec=0;
    fd_set readfd;
    FD_ZERO(&readfd);
    FD_SET(sockfd,&readfd);
    select(sockfd+1, &readfd, NULL, NULL, &timeout);
    int ret=0;
    if (FD_ISSET(sockfd, &readfd)) {
	ret=0;
    } else {
	ret=1;
    }
    return ret;
}

